package com.qf.cx.simplenms.topic.impl;

import java.util.Observable;
import java.util.Observer;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.qf.cx.simplenms.common.utils.SetUtils;
import com.qf.cx.simplenms.readergroup.api.ReaderGroup;
import com.qf.cx.simplenms.readergroup.tag.ReaderGroupTags;
import com.qf.cx.simplenms.report.model.Report;
import com.qf.cx.simplenms.tagmemory.model.TagMemoryModel;
import com.qf.cx.simplenms.topic.api.Topic;
import com.qf.cx.simplenms.topic.model.TopicModel;
import com.qf.cx.simplenms.topic.spi.TopicSpi;
import com.qf.cx.simplenms.topic.tag.TopicTagGenerator;
import com.qf.cx.simplenms.topic.util.ReportGenerator;

public class TopicImpl extends Observable implements Topic, TopicSpi, Runnable {
	private Logger logger = LoggerFactory.getLogger(this.getClass());

	private volatile TopicTagGenerator topicTagGenerator;
	private final TopicModel topicModel;

	private final AtomicBoolean started = new AtomicBoolean(false);
	private Thread reportGenerator;

	public TopicImpl(TopicModel topicModel) {
		this.topicModel = topicModel;
		this.topicTagGenerator = new TopicTagGenerator(topicModel.getTagMemories());
	}

	@Override
	public void start(Observer observer) {
		super.addObserver(observer);
		if (!started.compareAndSet(false, true)) {
			logger.debug("topic '{}' already started", getTopicName());
			return;
		}

		start0();
	}

	private void start0() {
		logger.info("topic '{}' starting", getTopicName());

		for (ReaderGroup t : topicModel.getReaderGroups()) {
			t.addTagMemories(TopicImpl.this, topicModel.getTagMemories());
			startReaderGroup(t);
		}

		reportGenerator = new Thread(this, this.getTopicName() + "_ReportsGenerator");
		reportGenerator.start();
		logger.info("topic '{}' started", getTopicName());
		if (logger.isDebugEnabled()) {
			logger.debug("ReaderGroup:");
			topicModel.getReaderGroups().forEach(t -> logger.debug(" " + t));
			logger.debug("TagMemory:");
			topicModel.getTagMemories().forEach(t -> logger.debug(" " + t));
		}
	}

	@Override
	public void stop(Observer observer) {
		super.deleteObserver(observer);
		if (countObservers() == 0 && started.compareAndSet(true, false)) {
			stop0();
		}
	}

	@Override
	public void forceStop() {
		stop0();
	}

	private void stop0() {
		logger.info("topic '{}' stoping", getTopicName());
		reportGenerator.interrupt();
		topicModel.getReaderGroups().forEach(t -> {
			t.deleteTagMemories(TopicImpl.this, topicModel.getTagMemories());
			stopReaderGroup(t);
		});
		deleteObservers();
		logger.info("topic '{}' stopped", getTopicName());
	}

	@Override
	public void update(Observable o, Object arg) {
		if (!started.get()) {
			return;
		}
		if (!(arg instanceof ReaderGroupTags)) {
			logger.error("can not handle type {}", arg.getClass());
			return;
		}
		ReaderGroupTags readerGroupTags = (ReaderGroupTags) arg;
		logger.debug("receive {}", readerGroupTags);
		topicTagGenerator.add(readerGroupTags);
	}

	@Override
	public TopicModel getTopicModel() {
		return topicModel;
	}

	@Override
	public void updateTopicModel(TopicModel newTopicModel) {
		synchronized (topicModel) {
			topicModel.setCycle(newTopicModel.getCycle());
			updateReaderGroups(newTopicModel.getReaderGroups());
			updateTagMemories(newTopicModel.getTagMemories());
			this.topicTagGenerator = new TopicTagGenerator(topicModel.getTagMemories());
		}
	}

	private void updateReaderGroups(Set<ReaderGroup> news) {
		Set<ReaderGroup> olds = topicModel.getReaderGroups();

		Set<ReaderGroup> dels = SetUtils.getComplementary(olds, news);
		deleteReaderGroups(dels);

		Set<ReaderGroup> adds = SetUtils.getComplementary(news, olds);
		addReaderGroups(adds);
	}

	private void deleteReaderGroups(Set<ReaderGroup> deletes) {
		if (!deletes.isEmpty()) {
			logger.info("topic '{}' delete reader group {}", getTopicName(), deletes);
		}
		if (started.get()) {
			for (ReaderGroup rg : deletes) {
				rg.deleteTagMemories(this, topicModel.getTagMemories());
				stopReaderGroup(rg);
			}
		}
		this.topicModel.deleteReaderGroups(deletes);

	}

	private void addReaderGroups(Set<ReaderGroup> news) {
		if (!news.isEmpty()) {
			logger.info("topic '{}' add reader group {}", getTopicName(), news);
		}
		if (started.get()) {
			for (ReaderGroup t : news) {
				t.addTagMemories(TopicImpl.this, topicModel.getTagMemories());
				startReaderGroup(t);
			}
		}
		this.topicModel.addReaderGroups(news);
	}

	private void updateTagMemories(Set<TagMemoryModel> news) {
		Set<TagMemoryModel> olds = topicModel.getTagMemories();

		Set<TagMemoryModel> dels = SetUtils.getComplementary(olds, news);
		deleteTagMemories(dels);

		Set<TagMemoryModel> adds = SetUtils.getComplementary(news, olds);
		addTagMemories(adds);
	}

	private void deleteTagMemories(Set<TagMemoryModel> deletes) {
		if (deletes.isEmpty()) {
			return;
		}
		logger.info("topic '{}' delete tms '{}'", getTopicName(), deletes);
		if (started.get()) {
			topicModel.getReaderGroups().forEach(t -> t.deleteTagMemories(TopicImpl.this, deletes));
		}

		topicModel.deleteTagMemories(deletes);
	}

	private void addTagMemories(Set<TagMemoryModel> adds) {
		if (!adds.isEmpty()) {
			logger.info("topic '{}' add tms '{}'", getTopicName(), adds);
		}
		if (started.get()) {
			for (ReaderGroup t : topicModel.getReaderGroups()) {
				t.addTagMemories(TopicImpl.this, adds);
				startReaderGroup(t);
			}
		}

		topicModel.addTagMemories(adds);
	}

	private void startReaderGroup(ReaderGroup t) {
		t.start(this);
	}

	private void stopReaderGroup(ReaderGroup t) {
		t.stop(this);
	}

	private String getTopicName() {
		return topicModel.getName();
	}

	@Override
	public void run() {
		for (;;) {
			try {
				TimeUnit.MILLISECONDS.sleep(this.topicModel.getCycle());
			} catch (InterruptedException e) {
				logger.warn("'{}' stopped", Thread.currentThread().getName());
				return;
			}

			Report report = ReportGenerator.generate(getTopicName(), topicTagGenerator.generate());
			if (!report.getBody().isEmpty()) {
				setChanged();
				notifyObservers(report);
			}
		}
	}

	@Override
	public String toString() {
		return "TopicImpl [topicModel=" + topicModel + ", started=" + started + "]";
	}

	@Override
	public String dump() {
		StringBuilder sb = new StringBuilder();
		sb.append("started=").append(started);

		sb.append("</br>Tag Memories:[");
		for (TagMemoryModel e1 : topicModel.getTagMemories()) {
			sb.append("[");
			sb.append(e1);
			sb.append("]");
		}
		sb.append("]");

		sb.append("</br>readerGroups:[");
		for (ReaderGroup e1 : topicModel.getReaderGroups()) {
			sb.append("[");
			sb.append(e1.getName());
			sb.append("]");
		}
		sb.append("]</br>");

		return sb.toString();
	}

	@Override
	public String getName() {
		return topicModel.getName();
	}
}
